package com.jzo2o.foundations.handler;


import com.jzo2o.canal.listeners.AbstractCanalRabbitMqMsgListener;
import com.jzo2o.es.core.ElasticSearchTemplate;
import com.jzo2o.foundations.constants.IndexConstants;
import com.jzo2o.foundations.model.domain.ServeSync;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;

/**
 * 接受从Canal发送的MQ消息并解析处理完成ES的索引更新
 * 服务数据同步处理器
 * server_sync表中数据变更生效，即：运营人员对服务信息的操作
 */
@Component
@Slf4j
public class ServeCanalDataSyncHandler extends AbstractCanalRabbitMqMsgListener<ServeSync> {

    @Resource
    private ElasticSearchTemplate esTemplate; // ES操作模板

    /**
     * 监听MQ的消息
     */
    //SpringAMQP的核心注解：@QueueBinding 声明队列、交换机及绑定关系，如果队列不存在则创建
    //                   @RabbitListener 监听MQ消息
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "canal-mq-jzo2o-foundations",  // 队列名称
                    arguments = @Argument(name = "x-single-active-consumer", value = "true", type = "java.lang.Boolean")), // 限制单消费者保证顺序性
            exchange = @Exchange(name = "exchange.canal-jzo2o", type = ExchangeTypes.TOPIC),   // 交换机
            key = "canal-mq-jzo2o-foundations"),   // Routing key
            // 限制单进程消费保证顺序性
            concurrency = "1")
    public void onMessage(Message message) throws Exception {
        // 调用抽象类中的方法：解析MQ消息并处理数据变更，解析消息后才会调用batchSave或batchDelete方法
        parseMsg(message);
    }


    /**
     * 解析到binlog中的新增、更新消息执行此方法，向es中保存数据
     *
     * @param data 待保存数据列表
     */
    @Override
    public void batchSave(List<ServeSync> data) {
        // 向es中保存数据
        Boolean aBoolean = esTemplate.opsForDoc().batchInsert(IndexConstants.SERVE, data);
        // 如果执行失败，要抛出异常，给MQ回nack消息。nack是在未到达
        if (!aBoolean) {
            try {
                Thread.sleep(1000); // 延迟1秒，避免重试机制导致死循环
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            throw new RuntimeException("同步失败");
        }
    }

    /**
     * 解析到binlog中的delete消息，将es中指定的文档进行删除
     *
     * @param ids 待删除数据ID列表
     */
    @Override
    public void batchDelete(List<Long> ids) {
        Boolean aBoolean = esTemplate.opsForDoc().batchDelete(IndexConstants.SERVE, ids);
        // 如果执行失败，要抛出异常，给MQ回nack消息
        if (!aBoolean) {
            try {
                Thread.sleep(1000); // 延迟1秒，避免重试机制导致死循环
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            throw new RuntimeException("同步失败");
        }
    }
}
